// Copyright 2023 Redpanda Data, Inc.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.md
//
// As of the Change Date specified in that file, in accordance with
// the Business Source License, use of this software will be governed
// by the Apache License, Version 2.0

#include "bytes/iobuf.h"
#include "cluster/self_test_rpc_types.h"

#include <seastar/testing/thread_test_case.hh>

SEASTAR_THREAD_TEST_CASE(test_make_netcheck_request) {
    static const auto check_frags = [](iobuf& buf, size_t sz) {
        vassert(
          std::distance(buf.begin(), buf.end()) >= 2,
          "check frags verifies constant fragment legnth up until final "
          "fragment in iobuf");
        auto it = buf.rbegin();
        ++it; // Its expected for the final fragment to have a size < 'sz'
        for (; it != buf.rend(); ++it) {
            if (it->size() != sz) {
                return false;
            }
        }
        return true;
    };

    auto req = cluster::make_netcheck_request(model::node_id{0}, 52).get().buf;
    BOOST_CHECK_EQUAL(req.size_bytes(), 52);
    BOOST_CHECK_EQUAL(std::distance(req.cbegin(), req.cend()), 1);
    req = cluster::make_netcheck_request(model::node_id{0}, 8192).get().buf;
    BOOST_CHECK_EQUAL(req.size_bytes(), 8192);
    BOOST_CHECK_EQUAL(std::distance(req.cbegin(), req.cend()), 1);
    req = cluster::make_netcheck_request(model::node_id{0}, 8193).get().buf;
    BOOST_CHECK_EQUAL(req.size_bytes(), 8193);
    BOOST_CHECK_EQUAL(std::distance(req.cbegin(), req.cend()), 2);
    /// Verify the first of 2 fragments has the expected size of 8192 while
    /// remainder leftover size must be in second fragment
    BOOST_CHECK_EQUAL(req.cbegin()->size(), 8192);
    BOOST_CHECK_EQUAL((++req.cbegin())->size(), 1);
    const auto size = ((8192 * 52) + 100);
    req = cluster::make_netcheck_request(model::node_id{0}, size).get().buf;
    BOOST_CHECK_EQUAL(req.size_bytes(), size);
    BOOST_CHECK_EQUAL(std::distance(req.cbegin(), req.cend()), 53);
    BOOST_CHECK(check_frags(req, 8192));
    BOOST_CHECK_EQUAL(req.rbegin()->size(), 100);
}
